package com.yanggu.flink.datastream_api.sink

import cn.hutool.core.date.DateUtil
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

import java.util.Date

object MySinkFunctionDemo {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment

    environment
      .fromCollection(List(1, 2, 3, 4))
      .addSink(new RichSinkFunction[Int] {

        /**
         *
         * @param value 对应的值
         * @param context 上下文信息
         */
        override def invoke(value: Int, context: SinkFunction.Context): Unit = {
          println(s"slot: ${getRuntimeContext.getTaskNameWithSubtasks}, 运行时间: ${DateUtil.formatDateTime(new Date(context.currentProcessingTime()))}" +
            s", 对应的值: $value")
        }
      })

    environment.execute("SinkDemo01 Job")
  }

}
